package com.yanggu.flink.datastream_api.multi_stream_transform.combine_stream

import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.functions.JoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * key相同且都在同一个窗口中数据，会按照笛卡尔积的形式输出
 * M X N的形式输出
 * 输出示例
 * (a,1000)=>(a,3000)
 * (a,1000)=>(a,4000)
 * (a,2000)=>(a,3000)
 * (a,2000)=>(a,4000)
 * (b,1000)=>(b,3000)
 * (b,1000)=>(b,4000)
 * (b,2000)=>(b,3000)
 * (b,2000)=>(b,4000)
 *
 * 第一条流中的("c", 3000L)并没有输出, 这是因为在第二条流中并没有在相同窗口内且为"c"的数据
 * 这是一种INNER JOIN的形式, 只有两条流在相同的窗口内都有数据才会输出
 */
object WindowJoinDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val dataStream1 = env
      .fromElements(
        ("a", 1000L),
        ("b", 1000L),
        ("a", 2000L),
        ("b", 2000L),
        ("c", 3000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
          override def extractTimestamp(element: (String, Long), recordTimestamp: Long) = element._2
        })
      )

    val dataStream2 = env
      .fromElements(
        ("a", 3000L),
        ("b", 3000L),
        ("a", 4000L),
        ("b", 4000L)
      )
      .assignTimestampsAndWatermarks(WatermarkStrategy
        .forMonotonousTimestamps()
        .withTimestampAssigner(new SerializableTimestampAssigner[(String, Long)] {
          override def extractTimestamp(element: (String, Long), recordTimestamp: Long) = element._2
        })
      )

    dataStream1
      .join(dataStream2)
      .where(tuple => tuple._1)
      .equalTo(tuple => tuple._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(5L)))
      .apply(new JoinFunction[(String, Long), (String, Long), String] {
        override def join(first: (String, Long), second: (String, Long)) = {
          first + "=>" + second
        }
      })
      .print()

    env.execute()

  }

}
